package com.study.flink.scala.day05_state

import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object StateBackendDemo {

  def main(args: Array[String]): Unit = {

    System.setProperty("HADOOP_USER_NAME", "root")

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val checkpointPath: String = "hdfs://hdp03:8020/flink/state/rocksdb"

    val backend = new RocksDBStateBackend(checkpointPath, true)
    backend.setDbStoragePath("file:///D:\\temp")

    env.setStateBackend(backend)
    env.enableCheckpointing(10000L)

    import org.apache.flink.api.scala._

    env.socketTextStream("flink-node01", 8888)
      .flatMap(_.split("\\s+"))
      .filter(_.trim.nonEmpty)
      .map((_, 1))
      .keyBy(0)
      .sum(1)
      .print("KeyedState--->")

    env.execute("StateBackendDemo")

  }

}
